package org.qiyu.live.user.provider.config;


import com.alibaba.fastjson.JSON;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.qiyu.live.user.interfaces.dto.UserDTO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import qiyu.live.framework.redis.starter.key.UserProviderCacheKeyBuilder;

import java.util.List;

/**
 * 消息消费者类
 */

@Configuration
public class RocketMQConsumerConfig implements InitializingBean {

    private final static Logger LOGGER = LoggerFactory.getLogger(RocketMQConsumerConfig.class);

    @Resource
    private RocketMQConsumerProperties consumerProperties;

    @Resource
    private RedisTemplate<String,UserDTO> softwareRedisTemplate;

    @Resource
    private UserProviderCacheKeyBuilder cacheKeyBuilder;

    @Override
    public void afterPropertiesSet() throws Exception {
        initConsumer();
    }

    private void initConsumer() {

        try {
            DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
            //设置namesrv地址
            defaultMQPushConsumer.setNamesrvAddr(consumerProperties.getNameSrv());
            //设置消费者组
            defaultMQPushConsumer.setConsumerGroup(consumerProperties.getGroupName());
            //设置最大拉取多少条消息
            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(1);
            //从哪里开始消费消息
            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //订阅的主题
            defaultMQPushConsumer.subscribe("user-update-cache", "*");
            defaultMQPushConsumer.setMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msg, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    LOGGER.info("收到消息。。。。。。。。。。。。。。。。。。。。。。");

                    String msgStr  = new String(msg.get(0).getBody());
                    UserDTO userDTO = JSON.parseObject(msgStr, UserDTO.class);
                    if(userDTO == null ||userDTO.getUserId()==null){
                        LOGGER.error("用户id为空，参数异常，内容:{}", msgStr);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    //延迟删除
                    String key = cacheKeyBuilder.buildUserInfoKey(userDTO.getUserId());
                    softwareRedisTemplate.delete(key);
                    LOGGER.info("延迟删除成功，userdto is {}",userDTO);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });


            defaultMQPushConsumer.start();
            LOGGER.info("mq消费者启动成功,nameSrv is {}", consumerProperties.getNameSrv());


        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }


    }
}
